%%--- coding:utf-8 ---
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% File Name: mongodb_driver
%%% Created on : 2024/4/16 20:39
%%% @author Gaylen 252323463@qq.com
%%% @copyright (C) 2024, freedom
%%% @doc
%%%
%%% @end
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-ifndef('mongodb_driver_HRL').
-define('mongodb_driver_HRL', true).

-define(GS2_HEADER, <<"n,,">>).
-define(NOT_MASTER_ERROR, 13435).
-define(UNAUTHORIZED_ERROR(C), C =:= 10057; C =:= 16550).
-define(ETS_MONGO_ID, ets_mongo_id).
-define(MAX_INT32, 2147483647).

-define(MONGO_DB_MODE_SINGLE, 0).           %% 单入口模式
-define(MONGO_DB_MODE_MASTER_SLAVE, 1).     %% 主从副本集模式

-define(MONGO_REQ_ASYNC, 0).    %% 默认进程外请求
-define(MONGO_REQ_PING, 1).     %% mc_worker进程内的ping请求
-define(MONGO_REQ_ISMASTER, 2). %% mc_worker进程内的is_master请求

-define(MONGO_OP_REPLY, 1).             %% MongoDB 5.0后弃用
-define(MONGO_OP_UPDATE, 2001).         %% MongoDB 5.0后弃用
-define(MONGO_OP_INSERT, 2002).         %% MongoDB 5.0后弃用
-define(MONGO_RESERVED, 2003).          %% 之前曾供 OP_GET_BY_OID 使用
-define(MONGO_OP_QUERY, 2004).          %% MongoDB 5.0后弃用
-define(MONGO_OP_GET_MORE, 2005).       %% MongoDB 5.0后弃用
-define(MONGO_OP_DELETE, 2006).         %% MongoDB 5.0后弃用
-define(MONGO_OP_KILL_CURSORS, 2007).   %% MongoDB 5.0后弃用
-define(MONGO_OP_MSG, 2013).            %% MongoDB 3.4版本后增加
-define(MONGO_OP_COMPRESSED, 2012).     %% MongoDB 3.4版本后增加

-type collection() :: binary() | atom(). % without db prefix
-type database() :: binary() | atom().
-type colldb() :: collection() | {database(), collection()}.
-type cursorid() :: integer().
-type selector() :: map() | bson:document().
-type projector() :: bson:document() | map().
-type skip() :: integer().
-type batchsize() :: integer(). % 0 = default batch size. negative closes cursor
-type modifier() :: bson:document() | map().
-type connection() :: pid().
-type args() :: [arg()].
-type arg() :: {database, database()}
| {login, binary()}
| {password, binary()}
| {w_mode, write_mode()}
| {r_mode, read_mode()}
| {host, list()}
| {port, integer()}
| {ssl, boolean()}
| {ssl_opts, proplists:proplist()}
| {register, atom() | fun()}.
-type write_mode() :: unsafe | safe | {safe, bson:document()}.
-type read_mode() :: master | slave_ok.
-type service() :: {Host :: inet:hostname() | inet:ip_address(), Post :: 0..65535}.
-type options() :: [option()].
-type option() :: {timeout, timeout()} | {ssl, boolean()} | ssl | {database, database()} | {read_mode, read_mode()} | {write_mode, write_mode()} | {host, string()} | {port, integer()}.
-type cursor() :: pid().
-type requestid() :: integer().

-record(mongo_worker_state, {
%%    conn_state :: conn_state(),     %% 连接的数据库信息和读写模式
    net_module :: ssl | gen_tcp,     %% socket的处理模块, gen_tcp|ssl
    socket = undefined :: gen_tcp:socket() | ssl:sslsocket() | undefined,   %% 数据库连接的socket
%%    is_master :: boolean(),         %% 此连接的节点是否主节点
    request_storage = #{} :: map(), %% 处理数据库的业务id集， key: 请示id, value: #mongo_reqstorage{}
    buffer = <<>> :: binary(),       %% 数据库返回的数据缓存
    alive_tick = 0 ::integer(),     %% 存活ping时间戳
    alive_timer = undefined :: reference() | undefined
}).

-record(mongo_reqstorage, {
    request_id :: integer(),
    req_from :: pid(),
    req_type = 0 :: integer()
}).

-record(mongo_cursor_state, {
    connection :: connection(),     %% 数据库进程mc_worker的进程id
    database :: database(),         %% 数据库名
    collection :: collection(),     %% 数据表
    cursor_id :: cursorid(),        %% 游标id
    batch_size :: integer(),        %% 每次读取多少数量的数据
    monitor :: pid()                %% 监控进程pid
}).

%% OP_MSG操作码对应的记录
-record(mongo_opmsg, {
    request_id = 0 :: requestid(),  %% 请求ID
    payload_type = 0 :: integer(),  %% 0 单个doc, 1 多个doc
    document :: map() | bson:document()
}).

%% OP_COMPRESSED操作码对应的记录
-record(mongo_opcompressed, {
    request_id = 0 :: requestid(),  %% 请求ID
    original_opcode = 0 :: integer(),   %% 包含封装操作码的值
    uncompressed_size = 0 :: integer(), %% 压缩后的compressedMessage字段的字节大小
    compressor_id = 0 :: integer(),     %% 压缩此消息的压缩器的 ID
                                        %% 0 此消息的内容未压缩
                                        %% 1 使用 snappy 进行压缩
                                        %% 2 使用 zlib 进行压缩
                                        %% 3 使用 zstd 进行压缩
    compressed_message :: binary()      %% 操作码本身，不包括 MsgHeader
}).

%% OP_REPLY操作码对应的记录
-record(mongo_opreply,  {
    request_id = 0 :: integer(),    %% 请求id起始值
    response_id = 0 :: integer(),   %% 回复对应的请求ID
    res_flag_cursor_not_found = false :: boolean(), %% 当调用了 getMore 但游标 ID 在服务器上无效时设置。返回结果为零
    res_flag_query_failure = false :: boolean(),    %% 在查询失败时设置。结果由一个包含描述失败的“$err”字段的文档组成
    res_flag_shard_config_stale = false :: boolean(),   %% 驱动程序应忽略这一点。只有 mongos 会看到此集合，在这种情况下，它需要从服务器更新配置。
    res_flag_await_capable = false :: boolean(),    %% 当服务器支持 AwaitData 查询选项时设置。如果没有，则客户端应在可追加游标的 getMore 之间休息一下。
    cursor_id = 0 :: integer(),     %% 游标ID, 64位整数, 如果查询结果只有一条OP_REPLY消息，则此字段为0
    starting_from = 0 :: integer(), %% 游标的起始位置
    number_returned = 0 :: integer(),   %% 回复中的文档数量
    documents :: [map() | bson:document()]
}).

%% OP_UPDATE操作码对应的记录
-record(mongo_opupdate, {
    request_id = 0 :: requestid(),  %% 请求ID
    zero = 0 :: integer(),          %% 整数值 0。保留供将来使用
    database :: database(),         %% 要更新的数据库
    collection :: colldb(),         %% 要更新的集合
    flags_upsert = false :: boolean(),  %% 设置此选项，则在未找到匹配的文档时，数据库会将提供的对象插入集合。
    flags_multiupdate = false :: boolean(), %% 如果设置此选项，数据库将更新集合中的所有匹配对象。否则，仅更新第一个匹配的文档。
    selector :: map() | bson:document(),    %% 更新记录的查询条件
    updater :: map() | bson:document()       %% 更新语句
}).

%% OP_INSERT操作码对应的记录
-record(mongo_opinsert, {
    request_id = 0 :: requestid(),  %% 请求ID
    flags_continue_on_error = false :: boolean(),   %% 如果设置此选项，则即使批量插入失败（如由于 ID 重复），数据库也不会停止处理批量插入。这使得批量插入的行为类似于一系列单次插入，只是任何插入失败而不仅仅是最后插入失败都会设置 lastError。如果出现多个错误，getLastError 只报告最新的错误。
    database :: database(),         %% 要插入的数据库
    collection :: colldb(),         %% 要插入的集合
    documents :: [map() | bson:document()]  %% 插入语句列表
}).

%% OP_QUERY操作码对应的记录, 运行 isMaster 和 hello 命令以作为连接握手的一部分仍然支持 OP_QUERY
-record(mongo_opquery, {
    request_id = 0 :: requestid(),  %% 请求ID
    flags_tailable_cursor = false :: boolean(), %% Tailable 表示检索最后一个数据时游标未关闭。相反，游标标记的是最终对象的位置。如果接收到更多数据，您可以稍后从游标所在的位置继续使用游标。与任何潜在游标一样，游标可能在某个时刻变得无效 (CursorNotFound) – 例如，如果它最终引用的对象被删除。
    flags_slave_ok = false :: boolean(),    %% 允许查询从属副本。通常，除了命名空间“local”之外，它们都会返回错误
    flags_oplog_replay = false :: boolean(),    %% 您无需指定此标志，因为 oplog 上符合条件的查询会自动进行优化。
    flags_no_cursor_timeout = false :: boolean(),   %% 服务器通常会在不活动期（10 分钟）后使空闲游标超时，防止过度使用内存。设置此选项，可防止出现这种情况
    flags_await_data = false :: boolean(),  %% 与 TailableCursor 一起使用。如果游标位于数据末尾，则会阻塞一段时间，而不是不返回数据。超时一段时间后，服务器正常返回
    flags_exhaust = false :: boolean(),     %% 假设客户端将完全读取所有查询的数据，则将以多个“更多”包的形式全速下传数据。当您拉取大量数据并知道要将其全部拉取时，速度会更快。注意：除非客户端关闭连接，否则不允许不读取所有数据。
    flags_part_read = false :: boolean(),  %% 如果某些分片不可用，则从 mongos 获取部分结果（而不是引发错误）
    database = undefined :: undefined | database(),         %% 要查询的数据库
    collection :: colldb(),         %% 要查询的集合
    number_to_skip = 0 :: integer(),    %% 设置返回查询结果时要忽略的文档数量（从结果数据集的第一个文档开始）
    number_to_return = 0 :: integer(),  %% 限制查询的第一条 OP_REPLY消息中的文档数量。 但是，如果结果多于numberToReturn ，数据库仍会建立游标并将cursorID返回给客户端
                                        %% 0，则数据库使用默认的返回大小
                                        %% 1，服务器会将该值视为 -1（自动关闭游标）
                                        %% 为负数时，数据库会返回该数字并关闭游标。无法获取该查询的其他结果
    selector :: map() | bson:document(),    %% 查询条件
    return_fields_selector = undefined :: undefined | map() | bson:document()   %% 可选。用于限制已返回文档中的字段的 BSON 文档。格式：{ a : 1, b : 1, c : 1}
}).

%% OP_GET_MORE操作码对应的记录
-record(mongo_opgetmore, {
    request_id = 0 :: requestid(),  %% 请求ID
    zero = 0 :: integer(),
    database :: database(),         %% 要查询的数据库
    collection :: colldb(),         %% 要查询的集合
    number_to_return = 0 :: integer(),  %% 设置返回的文档数量，与#mongo_opquery的number_to_return相同
    cursor_id :: integer()          %% 游标标识符， 64位整数
}).

%% OP_DELETE操作码对应的记录
-record(mongo_opdelete, {
    request_id = 0 :: requestid(),  %% 请求ID
    zero = 0 :: integer(),
    database :: database(),         %% 要查询的数据库
    collection :: colldb(),         %% 要查询的集合
    flags_single_remove = false :: boolean(),   %% 如果设置此选项，数据库将仅删除集合中第一个匹配的文档。否则，数据库将删除所有匹配的文档
    selector :: map() | bson:document()     %% 删除条件
}).

%% OP_KILL_CURSORS操作码对应的记录, 如果读取游标直至耗尽（读取直至 OP_QUERY 或 OP_GET_MORE 返回游标 ID 为零），则无需终止游标
-record(mongo_opkillcursors, {
    request_id = 0 :: requestid(),  %% 请求ID
    zero = 0 :: integer(),
    number_of_cursor_ids = 0 :: integer(),  %% 消息中的游标 ID 的数量
    cursor_ids = [] :: [integer()]  %% 要关闭的游标 ID 的“数组”。如果有多个操作码，则按顺序依次写入套接字
}).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% write
-record(insert, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    request_id :: integer(),
    documents :: [map() | bson:document()],
    bson_binary = <<>>  % insert编码后的二进制代码
}).

-record(update, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    request_id :: integer(),
    up_insert = false :: boolean(),
    multi_update = false :: boolean(),
    selector :: selector(),
    updater :: bson:document() | modifier(),
    bson_binary = <<>>  % update编码后的二进制代码
}).

-record(delete, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    request_id :: integer(),
    single_remove = false :: boolean(),
    selector :: selector(),
    bson_binary = <<>>  % delete编码后的二进制代码
}).

%% read
-record('query', {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    request_id :: integer(),
    tailablecursor = false :: boolean(),    %% Tailable 表示检索最后一个数据时游标不关闭
    slaveok = false :: boolean(),           %% 允许查询从属副本
    sok_overriden = false :: boolean(),     %% 删除，与slaveok重复
    nocursortimeout = false :: boolean(),   %% 对应于 NoCursorTimeout。服务器通常会在不活动期（10 分钟）后使空闲游标超时，防止过度使用内存。设置此选项，可防止出现这种情况
    awaitdata = false :: boolean(),         %% 与 TailableCursor 一起使用。如果游标位于数据末尾，则会阻塞一段时间，而不是不返回数据。超时一段时间后，服务器正常返回
    exhaust = false :: boolean(),           %% 假设客户端将完全读取所有查询的数据，则将以多个“更多”包的形式全速下传数据。当您拉取大量数据并知道要将其全部拉取时，速度会更快。注意：除非客户端关闭连接，否则不允许不读取所有数据。
    skip = 0 :: skip(),
    batchsize = 0 :: batchsize(),
    selector :: selector(),
    projector = #{} :: projector(),
    bson_binary = <<>>  % query编码后的二进制代码
}).
-type query() :: #query{}.

-record(getmore, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    request_id :: integer(),
    batchsize = 0 :: batchsize(),
    cursorid :: cursorid(),
    bson_binary = <<>>  % getmore编码后的二进制代码
}).

%% system
-record(ensure_index, {
    database :: database(),
    collection :: colldb(),
    request_id :: integer(),
    index_spec,
    bson_binary = <<>>  % ensure_index编码后的二进制代码
}).

%%-record(conn_state, {
%%    write_mode = unsafe :: write_mode(),
%%    read_mode = master :: read_mode(),
%%    database :: database()
%%}).
%%-type conn_state() :: #conn_state{}.

-record(killcursor, {
    cursorids :: [cursorid()]
}).

-record(reply, {
    request_id :: integer(),
    response_to :: integer(),
    flags_cursor_not_found :: boolean(),
    flags_query_error :: boolean(),
    flags_await_capable = false :: boolean(),
    cursor_id :: cursorid(),
    starting_from = 0 :: integer(),
    documents :: [map() | bson:document()]
}).
-type reply() :: #reply{}.

-record(mongo_manager_state, {
    worker_size :: integer(),       %% mc_worker连接进程数
    min_pool_size :: integer(),     %% 最小处理进程数
    max_pool_size :: integer() | unlimited, %% 最大处理进程数
    db_mode = ?MONGO_DB_MODE_SINGLE :: integer(),           %% 数据库模型: 0 单入口, 1 主从副本集
    worker_pid_list :: [pid()],     %% mc_worker连接进程id列表, TODO 检查并找出master的连接
    free_pool_list :: [pid()],      %% 空闲的处理进程id列表
    busy_pool_list :: [pid()],      %% 工作中的处理进程id列表
    org_options :: options(),       %% 原始连接参数
    options :: options(),           %% 主数据库节点的连接参数，每分钟检查一次主从，切换时断开所有连接，重新建立主连接
    worker_timer :: reference() | undefined,     %% 定时检查worker进程数
    free_pool_timer :: reference() | undefined     %% 定时检查并回收空闲的处理进程定时器
}).

-record(mongo_pool_state, {

}).

-record(mongo_insert, {
    database :: database(),     %% 要插入的数据库
    collection :: colldb(),     %% 要插入的数据表
    documents :: [map() | bson:document()]  %% 插入语句
}).

-record(mongo_query, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    selector :: map() | bson:document(),
    projector :: map(),
    skip :: integer(),
    limit :: integer()
}).

-record(mongo_update, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    selector :: map() | bson:document(),
    updater :: bson:document() | modifier(),
    up_insert = false :: boolean(),     %% 记录不存在时是否自动插入
    multi_update = false :: boolean()   %% 是否要更新多条记录
}).

-record(mongo_delete, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    selector :: map() | bson:document(),
    delete_one ::boolean()  %% 是否只删除一条记录
}).

-record(mongo_ensure_index, {
    database :: database(),  % overrides connection's database
    collection :: colldb(),
    index_specs :: bson:document(),     %% [{<<"字段名">>, 1升序|-1降序}, ...]
    unique :: boolean()     %% 是否唯一索引
}).

-record(mongo_batch_request, {
    request_from :: pid(),
    request_list :: [#mongo_insert{} | #mongo_query{} | #mongo_update{} | #mongo_delete{} | #mongo_ensure_index{}]
}).

% A notice is an asynchronous message sent to the server (no reply expected)
-type notice() :: #insert{} | #update{} | #delete{} | #killcursor{} | #ensure_index{}.
% A request is a synchronous message sent to the server (reply expected)
-type request() :: #'query'{} | #getmore{}.
-type message() :: notice() | request().

-define(WRITE(Req), is_record(Req, 'insert'); is_record(Req, 'update'); is_record(Req, 'delete')).
-define(READ(Req), is_record(Req, 'query'); is_record(Req, 'getmore')).


-endif.